package com.adcc.ags.utility.mq.transfer;
import com.adcc.ags.utility.log.Log;
import com.adcc.ags.utility.mq.configuration.MQConfiguration;
import com.adcc.ags.utility.mq.entity.AsyncReceiverState;
import com.adcc.ags.utility.mq.entity.Message;
import com.google.common.base.Optional;
import com.ibm.mq.*;
import com.ibm.mq.constants.CMQC;
import com.ibm.mq.headers.pcf.MQEPH;

import java.util.concurrent.Executors;

/**
 * 队列异步接收处理类
 */
public class AsyncQueueReceiver extends AsyncReceiver{

    // MQQueue
    private MQQueue mqq;

    // MQGetMessageOptions
    private MQGetMessageOptions gmo = new MQGetMessageOptions();

    /**
     * 构造函数
     */
    public AsyncQueueReceiver() {
        super();
    }

    /**
     * 构造函数
     * @param configuration
     * @param name
     */
    public AsyncQueueReceiver(MQConfiguration configuration, String name) {
        super(configuration, name);
    }

    /**
     * 构造函数
     * @param name
     * @param mqm
     * @param listener
     */
    public AsyncQueueReceiver(String name, MQQueueManager mqm, AsyncMsgListener listener) {
        super(name, mqm, listener);
    }

    /**
     * 取得队列深度
     * @return
     */
    private int getDepth(){
        try{
            if(mqq.isOpen()){
                return mqq.getCurrentDepth();
            }else {
                Log.error("queue:" + name + " is not open");
                state = AsyncReceiverState.Opening;
                return -1;
            }
        }catch (MQException ex){
            Log.error("getDepth() error",ex);
            state = AsyncReceiverState.Opening;
            return -1;
        }
    }

    /**
     * 接收消息
     * @return
     */
    private Optional<Message> receive(){
        try{
            if(mqq.isOpen()){
                MQMessage mqMessage = new MQMessage();
                mqMessage.characterSet = 1208;
                mqq.get(mqMessage, gmo);
                int msgLength = mqMessage.getMessageLength();
                byte[] buffer = new byte[msgLength];
                mqMessage.readFully(buffer,0,buffer.length);
                Message message = new Message(buffer);
                return Optional.of(message);
            }else{
                Log.error("queue:" + name + " is not open");
                state = AsyncReceiverState.Opening;
                return Optional.absent();
            }
        }catch (Exception ex){
            if(ex instanceof  MQException){
                MQException e = (MQException)ex;
                if(e.getReason() == CMQC.MQRC_NO_MSG_AVAILABLE || e.getReason() == CMQC.MQRC_OPTIONS_CHANGED){
                   return Optional.absent();
                }
            }
            Log.error("receive() error",ex);
            return Optional.absent();
        }
    }

    @Override
    public void open(){
        try{
            int options = CMQC.MQOO_INPUT_AS_Q_DEF | CMQC.MQOO_INQUIRE | CMQC.MQOO_FAIL_IF_QUIESCING;
            gmo.options = CMQC.MQGMO_NO_WAIT | CMQC.MQGMO_CONVERT | CMQC.MQGMO_FAIL_IF_QUIESCING;
            mqq = mqm.accessQueue(name,options);
            if(mqq.isOpen()){
                state = AsyncReceiverState.Open;
                executorService = Executors.newSingleThreadExecutor();
                executorService.execute(new Runnable() {
                    @Override
                    public void run() {
                        try{
                            while (true) {
                                try{
                                    if(getDepth() > 0 && state == AsyncReceiverState.Open){
                                        Optional<Message> optional = receive();
                                        if(optional.isPresent()){
                                            if(listener != null){
                                                listener.onMessage(configuration.getHost(),configuration.getPort(),configuration.getQueueManager(),name,optional.get(),0);
                                            }
                                        }
                                    }else{
                                        Thread.sleep(50);
                                        continue;
                                    }
                                }catch (Exception ex){
                                    Log.error("queue:" + name + " async receive message error",ex);
                                }
                            }
                        }catch (Exception ex){
                            Log.error("run() error",ex);
                        }
                    }
                });
            }else{
                state = AsyncReceiverState.Opening;
            }
        }catch (Exception ex){
            Log.error("open() error",ex);
            state = AsyncReceiverState.Opening;
        }
    }

    @Override
    public void close(){
        try{
            if(state != AsyncReceiverState.Closed){
                if(executorService != null && !executorService.isShutdown()){
                    executorService.shutdownNow();
                }
                state = AsyncReceiverState.Closed;
                mqq.close();
            }
        }catch (Exception ex){
            Log.error("close() error",ex);
        }
    }
}
